/*
 * JobServer.java
 *
 * Version:
 *      $Id: JobServer.java 19 2009-04-04 21:54:46Z PErickson87 $
 *
 * Copyright (c) 2009 Peter O. Erickson
 *
 * Permission is hereby granted, free of charge, to any person
 * obtaining a copy of this software and associated documentation
 * files (the "Software"), to deal in the Software without major
 * restriction, including without limitation the rights to use,
 * copy, modify, merge, publish, and/or distribute copies of, but
 * under no circumstances sublicense and/or sell, the Software,
 * and to permit persons to whom the Software is furnished to do
 * so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
 * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
 * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
 * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
 * OTHER DEALINGS IN THE SOFTWARE.
 */

package edu.rit.poe.cluster.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;
import java.util.Set;

/**
 * The JCluster Job Server acts likes an Air Traffic Controller, in that it
 * directs requesting jobs to specific nodes, and will monitor the overall state
 * of the cluster.
 * <p>
 * When <tt>Comm.init()</tt> is called in an application, it will connect to the
 * job server running on <tt>localhost</tt> and request nodes to run on.  The
 * job server will inform the job of the node hostnames to run on, and will only
 * do so when the nodes are available.
 * <p>
 * The job server also runs an HTTP interface to monitor the status of all
 * nodes.
 * 
 * @author  Peter O. Erickson
 *
 * @version $Id: JobServer.java 19 2009-04-04 21:54:46Z PErickson87 $
 */
public final class JobServer extends Thread {
    
    /** The well-known port for the job server. */
    public static final int DEFAULT_PORT = 4240;
    
    /** The default web server port for viewing the cluster. */
    public static final int DEFAULT_HTTP_PORT = 8085;
    
    /**
     * Enumerated type of potential node statuses.
     * 
     * @author  Peter O. Erickson
     * 
     * @version $Id: JobServer.java 19 2009-04-04 21:54:46Z PErickson87 $
     */
    public enum NodeStatus {
        
        /** Flag to indicate that a node is currently available. */
        AVAILABLE,
        
        /** Flag to indicate that a node is currently in use. */
        IN_USE;
        
    } // Status
    
    /** The job server configuration manager. */
    private ConfManager conf;
    
    /** Server socket for incoming job requests. */
    private ServerSocketChannel jobS;
    
    /** Server socket for incoming HTTP connections. */
    private ServerSocketChannel httpS;
    
    /** An array of the abstractions of nodes in the cluster. */
    private Node[] nodes;
    
    /** The number of nodes currently in use. */
    private int used;
    
    /** The total number of jobs scheduled by this service. */
    private int jobs;
    
    /** The queue for jobs waiting to be executed. */
    private Queue<Job> jobqueue;
    
    /** A 256 byte buffer for sending and receiving. */
    private ByteBuffer BUF;
    
    /** An ACSII character set decoder for received messages. */
    private CharsetDecoder DECODER;
    
    /**
     * Constructs a new job server with the default configuration.
     * 
     * @throws  IOException     if there is any I/O error with the nodes
     */
    public JobServer() throws IOException {
        this( ConfManager.DEFAULT_CONFIG_FILE );
    }
    
    /**
     * Constructs a new job server with the configuration from the specified
     * file.
     * 
     * @param   conffile        the configuration file name
     * 
     * @throws  IOException     if there is any I/O error with the nodes
     */
    public JobServer( String conffile ) throws IOException {
        conf = new ConfManager();
        conf.open( conffile );
        
        // 256 byte buffer
        BUF = ByteBuffer.allocateDirect( 256 );
        Charset charset = Charset.forName( "US-ASCII" );
        DECODER = charset.newDecoder();
        
        // a totally-ordering tree set, based on natural ordering
        nodes = new Node[ ( conf.getNodeHostnames() ).size() ];
        jobqueue = new LinkedList<Job>();
    }
    
    /**
     * Runs the job server until the application is terminated forcibly.
     */
    @Override
    public void run() {
        Selector selector = null;
        
        // setup the selector
        try {
            selector = Selector.open();
        } catch ( Exception e ) {
            System.err.println( "Error initializing selector." );
        }
        
        // setup the server socket channels
        try {
            jobS = ServerSocketChannel.open();
            jobS.socket().bind( new InetSocketAddress( conf.getJobPort() ) );
            jobS.configureBlocking( false );
            jobS.register( selector, SelectionKey.OP_ACCEPT );
            
            httpS = ServerSocketChannel.open();
            httpS.socket().bind( new InetSocketAddress( conf.getHTTPPort() ) );
            httpS.configureBlocking( false );
            httpS.register( selector, SelectionKey.OP_ACCEPT );
        } catch ( IOException e ) {
            System.err.println( "Error running server sockets." );
        }
        
        // connect to all nodes
        int i = 0;
        for ( String hostname : conf.getNodeHostnames() ) {
            SocketAddress addr =
                    new InetSocketAddress( hostname, conf.getNodePort() );
            SocketChannel socket = null;
            try {
                nodes[ i ] = new Node( hostname );
                
                socket = SocketChannel.open( addr );
                socket.configureBlocking( false );
                socket.register( selector, SelectionKey.OP_READ, nodes[ i ] );
            } catch ( IOException e ) {
                // TODO: crash the application nicely
            }
            i++;
        }
        
        // TODO: change from infini-loop
        while ( true ) {
            try {
                selector.select();
            } catch ( Exception e ) {
                // error selecting -- kill the loop
                break;
            }
            
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = keys.iterator();
            while ( iterator.hasNext() ) {
                
                // get the key, and remove it from the set
                SelectionKey key = ( SelectionKey ) iterator.next();
                iterator.remove();
                
                if ( key.isAcceptable() ) {
                    // accept new connection from to-be-pending job
                    SocketChannel channel = null;
                    try {
                        channel = jobS.accept();
                    } catch ( IOException e ) {
                        System.err.println(
                                "Error accepting client connection." );
                    }
                    System.out.println( "Connected to new client: "
                            + channel.socket().getInetAddress() );
                    
                    String[] args = null;
                    try {
                        // read the job information
                        args = read( channel );
                        
                        // message: job_name|#nodes
                        String jobname = args[ 0 ];
                        int n = Integer.parseInt( args[ 1 ] );
                        
                        // run or queue or throw an exception
                        // check for job that's too big
                        if ( n > conf.getClusteSize() ) {
                            throw new IllegalArgumentException( "Job requested "
                                    + "more nodes than cluster has." );
                        }
                        
                        if ( ( nodes.length - used ) >= n ) {
                            // enough nodes are available
                            Job job = new Job( jobname, channel, n );
                            
                            run( job );
                            
                        } else if ( ( nodes.length - used ) < n ) {
                            // not enough nodes available, queue the job
                            jobqueue.offer( new Job( jobname, channel, n ) );
                        }
                    } catch ( NumberFormatException e ) {
                        e.printStackTrace();
                        // error reading nodes request
                        try {
                            send( "ERR|Error reading nodes request: "
                                    + args[ 1 ], channel );
                        } catch ( IOException e2 ) {
                            // ignore
                        }
                    } catch ( IllegalArgumentException e ) {
                        // too many nodes requested
                        try {
                            send( "ERR|Too many nodes requested.", channel );
                        } catch ( IOException e2 ) {
                            // ignore
                        }
                    } catch ( ClosedChannelException e ) {
                        // ignore -- connection died
                    } catch ( CharacterCodingException e ) {
                        // error in the ASCII text -- ignore client
                    } catch ( IOException e ) {
                        // generic error reading -- ignore, nothing we can do
                    }
                } else if ( key.isReadable() ) {
                    SocketChannel channel = ( SocketChannel )key.channel();
                    // receive data from node
                    Node node = ( Node )key.attachment();
                    
                    String[] args = null;
                    try {
                        // read data from a node
                        args = read( channel );
                        
                        if ( args[ 0 ].equalsIgnoreCase( "DONE" ) ) {
                            node.status = NodeStatus.AVAILABLE;
                            node.job = null;
                            used--;
                            
                            // check for queued jobs
                            executeQueued();
                        }
                        
                    } catch ( IOException e ) {
                        // generic error reading -- ignore, nothing we can do
                    }
                }
            }
        }
    }
    
    /**
     * Runs the specified job on available nodes.
     * 
     * @param   job             the job to be run
     * 
     * @throws  IOException     if there is any communication error with the
     *                          nodes
     */
    private synchronized void run( Job job ) throws IOException {
        int n = job.nodes;
        // grab n nodes from the available set -- this is a dangerous loop, but
        // shouldn't ever look infinitely.  technically.  also, it would an
        // ArrayIndexOutOfBoundsException before looping indefinitely.  okay,
        // I'm going to stop writing notes to myself in my code.
        int assigned = 0;
        for ( int i = 0; assigned < n; i++ ) {
            Node node = nodes[ i ];
            if ( node.status == NodeStatus.AVAILABLE ) {
                node.status = NodeStatus.IN_USE;
                node.job = job;

                // send the hostname of this node to the
                // job requestor
                send( node.hostname + "|", job.channel );

                assigned++;
            }
        }
        used += n;
    }
    
    /**
     * Attempts to run any queued jobs in the order in which they were queued,
     * based on node availability.
     * 
     * @throws  IOException     if there is any communication error with the
     *                          nodes
     */
    private void executeQueued() throws IOException {
        if ( jobqueue.size() > 0 ) {
            Job job = jobqueue.peek();
            
            // are there enough nodes available to support the job?
            if ( job.nodes <= ( nodes.length - used ) ) {
                // remove from queue
                jobqueue.remove();
                
                // run the job
                run( job );
                
                // receusively check the job queue again
                executeQueued();
            }
        } 
    }
    
    /**
     * Send method to eliminate boilerplate code of buffer sending.
     * 
     * @param   asciiText       the ASCII text to be sent over the wire
     * @param   channel         the channel over which to send the text
     * 
     * @throws  IOException     if there is any error sending the text
     */
    private void send( String asciiText, SocketChannel channel )
            throws IOException {
        BUF.clear();
        BUF.put( asciiText.getBytes( "US-ASCII" ) );
        BUF.flip();
        channel.write( BUF );
    }
    
    /**
     * Read method to eliminate boilerplate code of buffer reading.
     * 
     * @param   channel         the channel to read data from
     * 
     * @return                  ASCII text from the channel
     * 
     * @throws  IOException     if there is any error reading the text
     */
    private String[] read( SocketChannel channel ) throws IOException {
        BUF.clear();
        channel.read( BUF );
        BUF.flip();
        return ( DECODER.decode( BUF ) ).toString().trim().split( "[|]" );
    }
    
    /**
     * Runs the main job server.
     * 
     * @param   args    the command line arguments
     */
    public static void main( String... args ) {
        // args[ 0 ]    hosts file
        
        JobServer jobserver = null;
        try {
            if ( args.length == 0 ) {
                jobserver = new JobServer();
            } else {
                jobserver = new JobServer( args[ 0 ] );
            }
        } catch ( IOException e ) {
            System.err.println(
                    "Error loading configuration file: " + e.getMessage() );
        }
        
        // run the job server
        jobserver.start();
    }
    
    /**
     * This class is an abstraction of a remote node in the cluster.
     * 
     * @author  Peter O. Erickson
     * 
     * @version $Id: JobServer.java 19 2009-04-04 21:54:46Z PErickson87 $
     */
    private final class Node {
        
        /** The hostname of the remote node. */
        private String hostname;
        
        /** The current status of the node. */
        private NodeStatus status;
        
        /** The job currently being executed on this node, if any. */
        private Job job;
        
        /**
         * Constructs a new node with the specified hostname.
         * 
         * @param   hostname        the hostname of the node
         */
        private Node( String hostname ) {
            this.hostname = hostname;
            this.status = NodeStatus.AVAILABLE;
        }
        
    } // Node
    
    /**
     * This class represents a job that is submitted to the job server.
     * 
     * @author  Peter O. Erickson
     * 
     * @version $Id: JobServer.java 19 2009-04-04 21:54:46Z PErickson87 $
     */
    private final class Job {
        
        /** The number of nodes requested for this job. */
        private int nodes;
        
        /** The name of this job. */
        private String jobname;
        
        /** The client socket channel that requested this job. */
        private SocketChannel channel;
        
        /**
         * Constructs a new job with the specified name and number of nodes.
         * 
         * @param   jobname     the name of the job
         * @param   channel     the channel that requested the job
         * @param   nodes       the number of nodes requested for the job
         */
        private Job( String jobname, SocketChannel channel, int nodes ) {
            this.jobname = jobname;
            this.channel = channel;
            this.nodes = nodes;
        }
        
    } // Job
    
} // JobServer
